package com.haoziqi.chapter_07;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * description
 * created by A on 2021/3/11
 */
public class Flink05_WaterMark {
    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 9999);
        SingleOutputStreamOperator<WaterSensor> d1 = ds.map(
                new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(
                                split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2])
                        );
                    }
                }
        )
                /* .assignTimestampsAndWatermarks(
                 WatermarkStrategy
                     .<WaterSensor>forMonotonousTimestamps()  //设置乱序时间,如果是升序的话不需要设置，因为升序默认吧乱序时间设置未来了0
                     .withTimestampAssigner(
                             new SerializableTimestampAssigner<WaterSensor>() {
                                 @Override
                                 public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                     return element.getTs()*1000L;
                                 }
                             }
                     )

         );
 */
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                //乱序时间
                                //水印生成策略: watermark = eventtime-乱序时间
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(10))//乱序时间
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<WaterSensor>() {
                                            @Override
                                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                                return element.getTs()*1000L;
                                            }
                                        }
                                ) //指定事件时间
                );
        //提交任务
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
